package com.example.ailiyuncloud.service.Impl;

import com.example.ailiyuncloud.service.KafkaService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;


/**
 * @Author: lqj
 * @Date: 2024/04/08/21:56
 * @Description:
 */
@Service
@Slf4j
public class KafkaServiceImpl implements KafkaService {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    public static final String TOPIC_TEST = "topic.test";
    public static final String TOPIC_GROUP1 = "topic.group1";

    //
    public static final String TOPIC_GROUP2 = "topic.group2";


    @Override
    public Object kafkaProducer(String topic, String info) {

//        ListenableFuture<SendResult<String, Object>> send = kafkaTemplate.send(topic, info);
//        send.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
//            @Override
//            public void onSuccess(SendResult<String, Object> stringObjectSendResult) {
//                //成功的处理
//                System.out.println("生产者 发送消息成功：" + stringObjectSendResult.toString());
//            }
//
//            @Override
//            public void onFailure(Throwable throwable) {
//                //发送失败的处理
//                System.out.println("生产者 发送消息失败：" + throwable.getMessage());
//            }
//
//        });
        return null;
    }

//    @Override
//    @KafkaListener(topics = "test", groupId = "hub-topic-city-01-group")
    public Object kafkaConsumer(ConsumerRecord<?, ?> record, Acknowledgment ack) {

//        try {
//            //1.从ConsumerRecord中获取消费数据
//            String originalMsg = (String) record.value();
//            log.info("从Kafka中消费的原始数据: " + originalMsg);
//            ack.acknowledge();
//            //2.把消费数据转换为DTO对象
//            //CityDTO cityDTO = JSONUtil.toBean(originalMsg, CityDTO.class);
//            //log.info("消费数据转换为DTO对象: " + cityDTO.toString());
//        } catch (Exception e) {
//            log.info("Consumer消费Topic异常",e);
//            e.printStackTrace();
//        }
        return null;
    }





}
